package com.atguigu.hadoop.examples.mapreduce;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.aggregate.ValueAggregatorBaseDescriptor;
import org.apache.hadoop.mapreduce.lib.aggregate.ValueAggregatorJob;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import java.util.StringTokenizer;

public class AggregateWordCount
{
    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException
    {
        Job job = ValueAggregatorJob.createValueAggregatorJob(args, new Class[] { WordCountPlugInClass.class });

        job.setJarByClass(AggregateWordCount.class);
        int ret = job.waitForCompletion(true) ? 0 : 1;
        System.exit(ret);
    }

    public static class WordCountPlugInClass extends ValueAggregatorBaseDescriptor
    {
        public ArrayList<Map.Entry<Text, Text>> generateKeyValPairs(Object key, Object val)
        {
            String countType = "LongValueSum";
            ArrayList retv = new ArrayList();
            String line = val.toString();
            StringTokenizer itr = new StringTokenizer(line);
            while (itr.hasMoreTokens()) {
                Map.Entry e = generateEntry(countType, itr.nextToken(), ONE);
                if (e != null) {
                    retv.add(e);
                }
            }
            return retv;
        }
    }
}